package com.orkasgb.framework.zookeeper.lock.api;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.text.NumberFormat;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;

/**
 *
 */
public class ZKApiLock {

    /**
     * 存放线程实例
     */
    private static final ThreadLocal<String> threadLocal = new ThreadLocal<>();

    /**
     * 模拟库存，库存增减必须保证原子性
     */
    private static final AtomicInteger count = new AtomicInteger(2);

    /**
     * 定义临时有序节点的父节点
     */
    private static final String locks = "/api_lock";

    /**
     * 主方法入口
     *
     * @param args 参数
     */
    public static void main(String[] args) {
        CountDownLatch countDownLatch = new CountDownLatch(5);
        /**
         * 定义5个线程，模拟客户端去对库存进行竞争操作。
         */
        for (int i = 0; i < 5; i++) {
            new Thread(() -> {
                try {
                    process(ZKApiClient.getZk());
                } catch (IOException | InterruptedException | KeeperException e) {
                    e.printStackTrace();
                }
            }, "thread" + i).start();
            countDownLatch.countDown();
        }
    }

    /**
     * 尝试加锁
     *
     * @param zk 客户端链接
     * @return 是否加锁成功
     */
    private static boolean tryLock(ZooKeeper zk) throws KeeperException, InterruptedException {
        try {
            /**
             * 支付的时候，必须保证线程安全。
             */
            if (lock(zk)) {
                System.out.println(Thread.currentThread().getName() + "-" + threadLocal.get() + "获取到了锁");
                pay();
                return true;
            }
        } finally {
            unlock(zk);
        }
            return false;

    }

    /**
     * 加锁逻辑
     *
     * @param zk
     * @return 是否加锁成功
     */
    private static boolean lock(ZooKeeper zk) {
        try {
            /**
             * 每个线程先去创建属于自己的临时顺序节点
             */
            String selfNode = zk
                    .create(locks + "/", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            /**
             * 保存线程自己创建的临时节点
             */
            threadLocal.set(selfNode);
            /**
             * 休眠一会，保证拿到了所有的线程创建的临时顺序节点
             */
            Thread.sleep(4);
            /**
             * 查询所有的节点信息
             */
            List<String> children = zk.getChildren(locks, false);
            /**
             * 对节点信息进行排序，排序号，第一个节点就是序号最小的
             */
            Collections.sort(children);
            System.out.println(Thread.currentThread().getName() + "创建节点为" + selfNode + "：所有的节点信息为" + children);
            String s = children.get(0);
            /**
             * 如果自己创建的临时顺序节点就是最小的那个节点，便获得锁
             */
            if (threadLocal.get().equals(locks + "/" + s)) {
                return true;
            }

            NumberFormat formatter = NumberFormat.getNumberInstance();
            formatter.setMinimumIntegerDigits(10);
            formatter.setGroupingUsed(false);
            int index = threadLocal.get().indexOf("/", 2) + 1;
            String self = threadLocal.get().substring(index, 16);
            /**
             * 如果自己创建的临时顺序节点不是最小的那个节点，便获取比他自己小1的那个节点，
             * 并监听那个节点的变化。
             */
            String prvNode = formatter.format(Integer.parseInt(self) - 1);
            final CountDownLatch countDownLatch = new CountDownLatch(1);
            /**
             * 监听比他自己小1那个节点的变化，当那个节点别删除的时候，比他自己小1那个节点
             * 就会通知他继续竞争锁。
             */
            Stat exists = zk.exists(locks + "/" + prvNode, watchedEvent -> {
                if (Watcher.Event.EventType.NodeDeleted.equals(watchedEvent.getType())) {
                    System.out.println(
                            "前一个节点 " + prvNode + " 被删除了 -" + Thread.currentThread().getName() + "被唤醒！唤醒节点信息为：" + self);
                    countDownLatch.countDown();
                }
            });

            /**
             * 如果存在比他自己小1那个节点，就等待那个节点被删除时通知他继续竞争锁。
             */
            if (exists != null) {
                System.out.println(Thread.currentThread().getName() + "-" + threadLocal.get() + "等待锁！开始监听" + prvNode + "节点变化");
                countDownLatch.await();
            }
        } catch (KeeperException | InterruptedException e) {
            e.printStackTrace();
        }

        return true;
    }

    /**
     * 解锁逻辑。删除自己创建的那个节点信息，删除自己创建的链接
     *
     * @param zk
     */
    private static void unlock(ZooKeeper zk) throws InterruptedException, KeeperException {
        zk.delete(String.valueOf(threadLocal.get()), -1);
        zk.close();
    }

    /**
     * 逻辑入口
     *
     * @param zk 客户端链接
     */
    static void process(ZooKeeper zk) {
        try {
            /**
             * 5个线程可以在不保证线程安全的情况下，去查询库存，
             * 有点类似抢票的时候，明明看到有余票，但是下单支付的时候，却提示无余票，支付失败。
             */
            System.out.println(Thread.currentThread().getName() + "查询库存：" + getInventory());
            tryLock(zk);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 获取当前剩余库存
     *
     * @return 剩余库存
     */
    private static int getInventory() {
        return count.get();
    }

    /**
     * 订单支付
     */
    private static void pay() {
        /**
         * 支付的时候，再次判断下是否还有余票。无余票则支付失败。
         */
        if (getInventory() == 0) {
            System.out.println(Thread.currentThread().getName() + "库存为0，支付失败.");
            return;
        }
        System.out.println(Thread.currentThread().getName() + "库存为，" + count + "支付成功");
        /**
         * 支付成功后，库存减1
         */
        count.decrementAndGet();
    }
}
